/* Copyright (C) 2008 Sahid Ferdjaoui <sahid@funraill.org>
 *
 *   This file is part of Postmemd.
 *
 * Postmemd is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */



#include "pm_head.h"
#include "pm_config.h"
#include "pm_slaves.h"
#include "pm_log.h"
#include "pm_poll.h"
#include "pm_atom.h"

#include <ev.h>
#include <json.h>

#include <stdio.h>

#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>



extern io_poller io;
extern config conf;

atom_t * atoms;

void
pm_slaves_process_loop (slave_t * slave)
{	
	int closed = 0, result;
	response_t response;

	while (!closed)
		{
			switch (slave->state)
				{
				case pm_read_master:
					result = pm_poll_read_fd (slave->read, & slave->rbuffer, & slave->rbuffer_bytes);
					switch (result)
						{
						case  0: 
							fprintf (stdout, "le slave à lu %s\n", slave->rbuffer);
							slave->state = pm_dispatch;
							slave->writed = 0;
							break;
						case  1: slave->state = pm_read_master; break;
						case -1: slave->state = pm_close; break;
						}
					break;
					
				case pm_dispatch:
					fprintf (stdout, "\n\n SLAVE DISPATCH \n\n");
					
					if (pm_slaves_parse_json (slave->rbuffer, & slave) != -1)
						{

							switch (slave->cmd)
								{
									/** remove data */
								case PM_CMD_DEL:
									result = pm_slave_del_atom (slave, & response);
									break;

									/** add atom in hash table	*/
								case PM_CMD_ADD:
								case PM_CMD_SET:
									result = pm_slave_add_atom (slave, & response);
									break;

								case PM_CMD_GET:
									result = pm_slave_get_atom (slave, & response);
									break;
								}
						}
					break;
					
				case pm_write_master:
					fprintf (stdout, "je dois ecrire : %s\n", slave->rbuffer);					

					result = pm_poll_write_fd (slave->write, slave->rbuffer, & slave->rbuffer_bytes, & slave->writed);
					switch (result)
						{
						case  0: slave->state = pm_close; fprintf (stdout, "le slave write %d\n", slave->rbuffer_bytes); break;
						case  1: slave->state = pm_write_master;  break;
						case -1: slave->state = pm_close; break;
						}
					break;
					
				case pm_close:
					fprintf (stdout, "le slave close\n");
					closed = 1;
					//pm_remove_fd (slave->loop, & slave->event);
					sleep (2);
					break;
				}
		}
	return;
}

void
pm_slaves_close (slave_t * slave)
{}

void
pm_slaves_read_fd (int fd, slave_t ** slave, char ** buffer, 
								 int * buffer_bytes, state_t next)
{
	int rbytes = 0;

	/** initialize read buffer */
	if (* buffer == NULL)
		{
			* buffer = malloc (conf.max_buffer_size);
			if (* buffer == NULL)
				{
					pm_log_error ("pm_slaves_read_fd () - malloc ()", PM_PRINT_ERRNO);
					sleep (1);
					return;
				}
			memset (* buffer, '\0', conf.max_buffer_size);
			* buffer_bytes = 0;
		}

	/** read file descriptor */
	rbytes = read (fd, * buffer + * buffer_bytes, conf.max_buffer_size - * buffer_bytes);
	if (rbytes > 0)
		{
			* buffer_bytes += rbytes;
			return;
		}
	if ((rbytes == 0) ||	((rbytes == -1) && (errno == EAGAIN)))
		{
			( * slave)->state = next;
			return;
		}
	pm_log_error ("pm_slaves_read_fd () - read ()", PM_PRINT_ERRNO);
	( * slave)->state = pm_close;
}


void
pm_slaves_init (pfd_t read_fd, pfd_t write_fd, struct ev_loop * loop)
{
	slave_t * slave = NULL;

	/** event loop */
	ev_default_fork ();

	/** set files descriptors no block */
	fcntl (read_fd, F_SETFL, fcntl (read_fd, F_GETFL) | O_NONBLOCK);

	slave = malloc (sizeof (slave_t));
	if (slave == NULL)
		{
			pm_log_error ("pm_slaves_init () - malloc ()", PM_PRINT_ERRNO);
			return;
		}

	slave->pid = getpid ();
	slave->read = read_fd;
	slave->write = write_fd;
	slave->loop = loop;

	slave->rbuffer = NULL;
	slave->rbuffer_bytes = 0;

	slave->state = pm_read_master;

	slave->event.data = slave;
	
	/** Record listener in event loop */
	pm_record_fd (loop, & slave->event, slave->read, pm_event_handler_slaves, EV_READ); 

	ev_loop (loop, 0);
	
	exit (0);
}


	
int
pm_slaves_parse_json (char * buffer, slave_t ** slave)
{
	struct json_object	  * json_cmd       = NULL,
		                    * json_key       = NULL;
	char			              cmd_string[10];

	/** parse json message */
	( * slave)->json_obj = json_tokener_parse (buffer);
	if (!( * slave)->json_obj)
		{
			pm_log_error ("message format is not json format", PM_NOPRINT_ERRNO);
			return -1;
		}
	
	/** parse json command */
	json_cmd = json_object_object_get (( * slave)->json_obj, PM_MSG_COMMAND);
	if (is_error (json_cmd))
		{
			return -1;
		}
	strcpy (cmd_string, json_object_get_string (json_cmd));
	
	if ((strcmp (cmd_string, PM_STR_CMD_ADD) == 0 && (( * slave)->cmd = PM_CMD_ADD)) ||
			(strcmp (cmd_string, PM_STR_CMD_SET) == 0 && (( * slave)->cmd = PM_CMD_SET)) ||
			(strcmp (cmd_string, PM_STR_CMD_DEL) == 0 && (( * slave)->cmd = PM_CMD_DEL)) ||
			(strcmp (cmd_string, PM_STR_CMD_GET) == 0 && (( * slave)->cmd = PM_CMD_GET)))
		{
			json_key = json_object_object_get (( * slave)->json_obj, PM_MSG_DATA_KEY);
			if (is_error (json_key))
				{
					return -1;
				}
			strncpy (( * slave)->key, json_object_get_string (json_key), 41);	
			
			return 0;
		}
	return -1;
}






int pm_slave_del_atom (slave_t * slave, response_t * response)
{
	strcpy (response->commande, PM_STR_CMD_DEL);
	strcpy (response->key, slave->key);
	
	if (pm_atom_del (slave->key) == -1)
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}
	
	strcpy (response->status, PM_RESPONSE_SUCCESS);
	return -1;
}

int pm_slave_get_atom (slave_t * slave, response_t * response)
{
	atom_t * atom = NULL;
	response->data = NULL;

	strcpy (response->commande, PM_STR_CMD_GET);
	strcpy (response->key, slave->key);

	if (pm_atom_get (slave->key, & atom) == -1)
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}

	response->data_bytes = atom->data_bytes;

	response->data = malloc (atom->data_bytes);
	if (response->data == NULL)
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}

	strcpy (response->data, atom->data);
	strcpy (response->status, PM_RESPONSE_SUCCESS);

	return 0;
}

int pm_slave_add_atom (slave_t * slave, response_t * response)
{
	atom_t * atom = NULL;

	struct json_object	  * json_data      = NULL, 
		                    * json_data_size = NULL,
		                    * json_expire    = NULL;

	strcpy (response->commande, PM_STR_CMD_GET);
	strcpy (response->key, slave->key);

	if (pm_atom_init (& atom) == -1)
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}

	strncpy (atom->key, slave->key, 41);

	/* expire atom */
	json_expire = json_object_object_get (slave->json_obj, PM_MSG_DATA_EXPIRE);
	if (is_error (json_expire))
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}
	atom->expire = json_object_get_int (json_expire);

	/* atom data size */
	json_data_size = json_object_object_get (slave->json_obj, PM_MSG_DATA_SIZE);
	if (is_error (json_data_size))
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}
	atom->data_bytes = json_object_get_int (json_data_size);

	/* atom data */
	json_data = json_object_object_get (slave->json_obj, PM_MSG_DATA);
	if (is_error (json_data))
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}
			
	atom->data = malloc (atom->data_bytes);
	if (atom->data == NULL)
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}
	strncpy (atom->data, 
					 json_object_get_string (json_data), atom->data_bytes);

	if (pm_atom_add (atom) == -1)
		{
			strcpy (response->status, PM_RESPONSE_FAILED);
			return -1;
		}

	strcpy (response->status, PM_RESPONSE_SUCCESS);
	return 0;

}



































int
pm_slaves_pool_init (pool_slaves_t ** pool)
{
	( * pool) = malloc (sizeof (pool_slaves_t));
	if (( * pool) == NULL)
		{
			pm_log_error ("pm_slaves_pool_init () - malloc ()", PM_PRINT_ERRNO);
			return -1;
 		}

	( * pool)->first = NULL;
	( * pool)->curr = NULL;
	( * pool)->last = NULL;

	( * pool)->size = 0;
	
	return 0;
}

int
pm_slaves_pool_add (pool_slaves_t ** pool, pid_t pid,	
										pfd_t read_fd, pfd_t write_fd, struct ev_loop * loop)
{
	slave_t * slave = NULL;
	
	slave = malloc (sizeof (slave_t));
	if (slave == NULL)
		{
			pm_log_error ("pm_slaves_pool_add () - malloc ()", PM_PRINT_ERRNO);
			return -1;
		}

	/** set files desriptors no block */
	fcntl (read_fd, F_SETFL, fcntl (read_fd, F_GETFL) | O_NONBLOCK);
	fcntl (write_fd, F_SETFL, fcntl (write_fd, F_GETFL) | O_NONBLOCK);

	slave->next = NULL;
	slave->prev = ( * pool)->last;

	slave->pid = pid;
	slave->read = read_fd;
	slave->write = write_fd;

	/** Record listener in event loop */
	//	pm_record_fd (loop, & io.slaves, io_event_select, pm_event_handler, EV_READ); 

	if (( * pool)->last)
		( * pool)->last->next = slave;

	( * pool)->curr = slave;
	( * pool)->last = slave;
	( * pool)->size ++;
	
	if (( * pool)->first == NULL)
		( * pool)->first = slave;

	return 0;
}

void
pm_slaves_pool_view (pool_slaves_t * pool)
{
	slave_t * slave;

	fprintf (stdout,"nombre de slaves : %i\n", pool->size);
	for (slave = pool->first; slave; slave = slave->next)
		fprintf (stdout, "slave pid: %i\n", slave->pid);
}

void 
pm_slaves_pool_change_curr (pool_slaves_t ** pool)
{
	( * pool)->curr = ( * pool)->curr->next;
	if (( * pool)->curr == NULL)
		( * pool)->curr = ( * pool)->first;
}

void
pm_slaves_pool_select_curr (pid_t pid, pool_slaves_t ** pool)
{
	int i;
	for (i = 0; i < conf.num_slaves; i ++)
		{
			pm_slaves_pool_change_curr (pool);
			if (( * pool)->curr->pid == pid)
				return;
		}
}





